package task

import (
	"database/sql"
	"sync"
	"EasyCollect/global"
	log "code.google.com/p/log4go"
	"EasyCollect/summary"
	"time"
	"EasyCollect/util"
	"EasyCollect/pool"
)

/**
	逻辑执行对象
 */
type UnitTask struct {
	SrcDb *sql.DB //原始库的数据库
	TargetDb *sql.DB //数据入库目标库的数据库连接
	Task     *global.Task      //任务配置
	workerPool *pool.Pool//提交任务的工作协程池引用。(未分片，使用这个)
	shardingWorkerPools map[int]*pool.Pool //提交任务的工作协程池引用。（分片的任务使用这个，一个分片单独一个池运行）
	Running  bool              //当前任务是否在执行
	lock     *sync.Mutex       //程序执行锁
	Pause    bool              //是否是暂停状态，如果状态是暂停，则逻辑不执行
}

/**
	逻辑执行单元，采用定时任务的方式启动
	如果task的isFlow配置为true，则只要能查询出数据，就会一直执行。
	如果是false，定时任务触发后只执行一次。
 */
func (t *UnitTask) cronRun(){
	if t.Task.IsFlow {
		for {
			size := t.waitRun()
			if size == 0 {
				break
			}
		}
	}else{
		t.waitRun()
	}
}

/**
	逻辑执行单元，采用人工触发的方式,人工触发的方式，只要能查询出数据，就会一直执行
	直到查询不出数据为止，该此触发的任务则为完成
	返回该次查询出的数据的大小
	有分片项的任务不支持流式任务和流式执行，只会在触发后执行一次
 */
func (t *UnitTask) waitRun() (size int){
	if !t.Pause {
		if !t.Running && t.setRunning() {

			defer t.setUnRunning()

			if t.Task.ShardingCount>0 {

				c := make(chan int)
				for i:=0;i< t.Task.ShardingCount; i++{
					go t.ShardingRun(i,c)
				}
				//在主线程阻塞，直到所有的分片的任务全部完成
				for i:=0;i< t.Task.ShardingCount; i++{
					<- c
				}
				log.Info("taskId [%s] taskName [%s] taskCommand [%s] excuseDown [%s]",t.Task.TaskId, t.Task.TaskName, t.Task.TaskCommand,
					util.GetCurrentTimeStr())
				return 0
			}
			return t.simpleRun()
		}
	}else{
		log.Info("taskId [%s] taskName [%s] status is stopped now",
			t.Task.TaskId,t.Task.TaskName)
	}
	return 0
}

/**
	没有配置分片，则执行此方法
 */
func (t *UnitTask) simpleRun() (size int){
	task := t.Task
	results := summary.FetchData(task, t.SrcDb,nil)
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] Database [%s] Table [%s] resultsSize [%v]",
		task.TaskId,task.TaskName, task.TaskCommand, task.SrcDatabase, task.SrcTable, len(results))


	err := summary.ExcuseTaskBeforeSql(task,t.TargetDb)
	if err ==nil {
		t.submitResultIntoWorker(results)
	}

	log.Info("taskId [%s] taskName [%s] taskCommand [%s] excuseDown [%s]",task.TaskId, task.TaskName, task.TaskCommand,
		util.GetCurrentTimeStr())

	time.Sleep(time.Duration(task.Speed)*time.Millisecond)
	return len(results)
}

/**
	配置分片，则执行此方法
	sharding：分片项的值
 */
func (t *UnitTask) ShardingRun(sharding int,c chan int){
	task := t.Task
	shardingItem := &summary.ShardingItem{Value:sharding}
	results := summary.FetchData(task, t.SrcDb,shardingItem)
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] sharding: [%v] Database [%s] Table [%s] resultsSize [%v]",
		task.TaskId,task.TaskName, task.TaskCommand,sharding,task.SrcDatabase, task.SrcTable,len(results))

	err := summary.ExcuseTaskBeforeSqlSharding(task,t.TargetDb,sharding)
	if err ==nil {
		t.submitResultIntoWorkerSharding(results,shardingItem.Value)
	}

	log.Info("taskId [%s] taskName [%s] taskCommand [%s] sharding: [%v] excuseDown [%s]",task.TaskId, task.TaskName, task.TaskCommand,
		sharding,util.GetCurrentTimeStr())
	c <- 0
}

/**
	将要处理的未分片的任务数据集仍进协程池
 */
func (t *UnitTask) submitResultIntoWorker(results []map[string]interface{}){

	for index := range results {
		result := results[index]
		t.workerPool.AddJob(func(){
			summary.ExcuseInsertOrUpdate(t.Task,t.SrcDb,t.TargetDb,result)
		})
	}
	t.workerPool.WaitForAll()
}


/**
	将要处理的需要分片的任务数据集仍进协程池。
 */
func (t *UnitTask) submitResultIntoWorkerSharding(results []map[string]interface{},sharding int){

	workerPool:= t.shardingWorkerPools[sharding]
	for index := range results {
		result := results[index]
		workerPool.AddJob(func(){
			summary.ExcuseInsertOrUpdate(t.Task,t.SrcDb,t.TargetDb,result)
		})
	}
	workerPool.WaitForAll()
}

/**
	对任务的一个描述
 */
func (t *UnitTask) describe(){

	task := t.Task

	if task.SrcDelete && task.TargetType==global.InsertType {
		log.Info("taskId [%s] taskName [%s] 任务类型：从mysql 表src-table中查询出数据,插入到target-table中。并删除src-table中的数据",task.TaskId,task.TaskName)
	}
	if !task.SrcDelete && task.TargetType==global.InsertType {
		log.Info("taskId [%s] taskName [%s] 任务类型：从mysql 表src-table中查询出数据,并插入到target-table中。不会删除src-table中的数据",task.TaskId,task.TaskName)
	}
	if  task.TargetType==global.UpdateType {
		log.Info("taskId [%s] taskName [%s] 任务类型：从mysql 表src-table中查询出数据,并根据相关字段update表target-table中的对应记录，不会删除src-table中的数据",task.TaskId,task.TaskName)
	}

}
/**
	将未执行状态改为执行状态
	如果程序正在执行，更改则失败。利用互斥锁实现，仅对当前结构体的多次调用生效
 */
func (t *UnitTask) setRunning() (success bool ){
	t.lock.Lock()
	if t.Running {
		return false
	}
	t.Running =true
	defer t.lock.Unlock()
	return true
}
/**
	将正在执行的任务的状态改为未执行
 */
func (t *UnitTask) setUnRunning() {
	t.Running =false
}
